use std::{cmp::min, fs, io::{Read, Write}, net::{TcpListener, TcpStream}, task::Poll, time::Duration};

use async_std::io::{ReadExt, WriteExt};
use futures::StreamExt;

async fn main_block() {
    println!("web server simple!!");
    // 监听当前机器的8888端口
    let listener = TcpListener::bind("127.0.0.1:8888").unwrap();
    // 阻塞等待请求的进入
    // listener.incoming 是阻塞的迭代器, 当listener在等待连接时,执行器时无法执行其他Future的,而且只有在我们处理完已有的连接后,才能接收新的连接.
    for stream in listener.incoming() {
        // 从Result解包得到TcpStream
        let tcpstream = stream.unwrap();
        // 经过各种async 和 此处的await，现在代码是异步的了，但是还无法并发。
        handle_connection(tcpstream).await;
    }
}

// 将上边mainblock中的阻塞的迭代器替换为非阻塞的Stream
#[async_std::main]
async fn main() {
    // 使用 async_std 包中的TcpListener 而不是 net 包中的.
    // 该TcpListener实现了Future,可以进行await这种异步处理.
    let listener = async_std::net::TcpListener::bind("127.0.0.1:8888").await.unwrap();
    // listener.incoming 不再阻塞, for_each_concurrent 可以并发地处理从Stream中获取地元素.
    // 有一个请求在处理中阻塞都会尝试再去获取另一个请求继续处理.
    listener.incoming().for_each_concurrent(None, |tcpstream| async {
        let tcpstream_real = tcpstream.unwrap();
        // 处理地过程中如果阻塞了,就把线程控制权让出来,不阻塞
        //handle_connection_async(tcpstream_real).await;
        // 使用多线程处理
        async_std::task::spawn(handle_connection_async(tcpstream_real));
    }).await;
}

// 处理连接...
// 此处没有使用借用，而是直接得到了所有权，也就是我们在全权处理这个链接。
// 如果改为 &mut TcpStream 还要维护生命周期，后续使用stream还必须确保借用不会冲突。
// 在当前场景下handle_connection处理完stream后就不再需要它了，因此直接转移所有权更简单更高效。
// + 所有权：此处直接使用的变量，而不是引用，所以所有权转移到了函数内部。
// + 可变性：默认情况下是不可变的，只有声明了mut才能改变。相当于在赋值的时候使用 let mut a = "hello".to_string();
// --------------
// 将handle_connection从普通函数修改为async函数
// 修改为async函数后，返回值就从()变为了 Future<Output=()>

async fn handle_connection(mut stream: TcpStream) {
    // 从连接中读取1024个字节 
    let mut buffer = [0; 1024];
    stream.read(&mut buffer).unwrap();

    // 根据请求行是不是 GET / HTTP/1.1\r\n 来决定是输出200的请求行还是404的请求行
    let (status_line, filename) = if buffer.starts_with(b"GET / HTTP/1.1\r\n") {
        ("HTTP/1.1 200 OK\r\n\r\n", "hello.html")
    } else if buffer.starts_with(b"GET /sleep HTTP/1.1\r\n") {
        // 请求 /sleep 页面
        // 此处模拟一次用户慢请求
        // 我们没有使用 std::thread::sleep 进行睡眠，因为该函数时阻塞的，它会让当前线程陷入睡眠中，导致其他任务无法继续执行。
        // 此处使用的 async_std::task::sleep 仅仅会让当前的任务陷入睡眠，然后该任务会让出线程的控制权，这样线程就可以继续运行其他的任务.
        async_std::task::sleep(Duration::from_secs(5)).await;
        ("HTTP/1.1 200 OK\r\n\r\n", "hello.html")
    } else {
        ("HTTP/1.1 404 NOT FOUND\r\n\r\n", "404.html")
    };
    // 基于filename读取内容
    let contents = fs::read_to_string(filename).unwrap();
    // 构造http响应格式：此处没有返回响应头
    // 状态行\r\n
    // 响应头\r\n
    // \r\n
    // 响应体
    let response = format!("{status_line}{contents}");
    // 将回复内容写入连接缓存中
    stream.write_all(response.as_bytes()).unwrap();
    // flush将缓存中的内容发送到客户端
    stream.flush().unwrap();
}

// 处理收到的每一个连接
// 该方法相比 handle_connection来说,在读取/写入/flush 三个步骤都使用了await的方式,在阻塞时让出线程的控制权,让别的任务执行
//async fn handle_connection_async(mut stream: async_std::net::TcpStream) {
    
// 为了可以使用单元测试,将 TcpStream 改为 impl async_std::io::Read/async_std::io::Write/marker::Unpin 
// 实际上 TcpStream 就是实现了这三个特征
async fn handle_connection_async(mut s: impl async_std::io::Read + async_std::io::Write + std::marker::Unpin) {
    let mut buffer = [0; 1024];
    s.read(&mut buffer).await.unwrap(); // 异步处理, 阻塞了也不影响其他的任务继续处理

     // 根据请求行是不是 GET / HTTP/1.1\r\n 来决定是输出200的请求行还是404的请求行
     let (status_line, filename) = if buffer.starts_with(b"GET / HTTP/1.1\r\n") {
        ("HTTP/1.1 200 OK\r\n\r\n", "hello.html")
    } else if buffer.starts_with(b"GET /sleep HTTP/1.1\r\n") {
        // 请求 /sleep 页面
        // 此处模拟一次用户慢请求
        // 我们没有使用 std::thread::sleep 进行睡眠，因为该函数时阻塞的，它会让当前线程陷入睡眠中，导致其他任务无法继续执行。
        // 此处使用的 async_std::task::sleep 仅仅会让当前的任务陷入睡眠，然后该任务会让出线程的控制权，这样线程就可以继续运行其他的任务.
        async_std::task::sleep(Duration::from_secs(5)).await;
        ("HTTP/1.1 200 OK\r\n\r\n", "hello.html")
    } else {
        ("HTTP/1.1 404 NOT FOUND\r\n\r\n", "404.html")
    };
    // 基于filename读取内容
    let contents = fs::read_to_string(filename).unwrap();
    // 构造http响应格式：此处没有返回响应头
    // 状态行\r\n
    // 响应头\r\n
    // \r\n
    // 响应体
    let response = format!("{status_line}{contents}");

    s.write(response.as_bytes()).await.unwrap();
    s.flush().await.unwrap();
}




// test ++++++++++++++++++++++++++
struct MockTcpStream {
    read_data: Vec<u8>,
    write_data: Vec<u8>,
}

// 实现Read特征
// 该实现将会包含一些read_data数据拷贝到buf缓存中,然后返回Poll::Ready说明read已经结束.
impl async_std::io::Read for MockTcpStream {
    fn poll_read(
        self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &mut [u8],
    ) -> Poll<std::io::Result<usize>> {
        // 将read_data中的数据拷贝到buf之中
        // 先获取buf和read_data二者的较小值,反正是不能超过buf的最大长度
        let size = min(self.read_data.len(), buf.len());
        // 拷贝到buf之中
        buf[..size].copy_from_slice(&self.read_data[..size]);
        Poll::Ready(Ok(size))
    }
}

impl async_std::io::Write for MockTcpStream {
    // 为该类型的可变引用显示的定义 poll_write 方法
    // 将buf中的数据写入到self.write_data之中
    fn poll_write(
        self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &[u8],
    ) -> Poll<std::io::Result<usize>> {
        //self.write_data = Vec::from(buf);
        // 安全地获取对 `self` 的可变引用
        let this = unsafe { self.get_unchecked_mut() };
        this.write_data = Vec::from(buf);
        Poll::Ready(Ok(buf.len()))
    }

    fn poll_flush(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<std::io::Result<()>> {
        Poll::Ready(Ok(()))
    }

    fn poll_close(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<std::io::Result<()>> {
        Poll::Ready(Ok(()))
    }
}

// MockTcpStream 实现Unpin特征,表示它可以在内存中安全的移动.
use std::marker::Unpin;
impl Unpin for MockTcpStream {}

// 单元测试用例 
#[async_std::test]
async fn test_handle_connection_async() {
    let mut contents = vec![0u8; 1024];

    // 将contents填充为基本的http请求行
    let input_bytes = b"GET / HTTP/1.1\r\n";
    contents[..input_bytes.len()].clone_from_slice(input_bytes);

    let mut stream = MockTcpStream { 
        read_data: contents, 
        write_data: Vec::new(),
    };

    // 异步处理
    handle_connection_async(&mut stream).await;

    // let mut buf = [0u8; 1024];
    // stream.read(&mut buf).await.unwrap();

    // 如果处理完毕了,理论上 stream.write_data 中应该已经有数据了
    let expected_contents = fs::read_to_string("hello.html").unwrap();
    let expected_response = format!("HTTP/1.1 200 OK\r\n\r\n{}", expected_contents);
    assert!(stream.write_data.starts_with(expected_response.as_bytes()));
}

// test --------------------------